import datetime
import decimal
import os
from asyncio import as_completed
from concurrent.futures import ThreadPoolExecutor

import numpy as np
import pandas as pd
from certifi.core import where
from dateutil.relativedelta import relativedelta
from retrying import retry
from tqdm import tqdm

import concurrent.futures

from components.config import WFQ_SOURCE_MYSQL_CONFIG
from utils.db.mysqldb import MysqlDB

RESULT_MALL_KEYS = [
    "version",
    "platform_id",
    "platform_name",
    "mall_id",
    "mall_name",
    "company_name_src",
    "taxpayer_id",
    "company_name",
    "month_sales",
    "month_sales_money",
    "total_sales",
    "total_sales_money",
    "percent_month_sales",
    "percent_month_sales_money",
    "percent_total_sales",
    "percent_total_sales_money",
    "goods_num",
    "self_support",
    "mall_url",
    "mall_open_time",
    "mall_open_time_year",
    "mall_status",
    "mall_end_time",
    "mall_platform_classification",
    "mall_business_classification",
    "mall_scope",
    "company_province",
    "company_city",
    "company_county",
    "company_town",
    "mall_location_code",
    "mall_send_province",
    "mall_send_city",
    "mall_send_county",
    "mall_send_location_code",
    "mall_desc",
    "mall_logo",
    "mall_star",
    "mall_label",
    "mall_rating_text",
    "mall_achievements_rank",
    "mall_review_num",
    "mall_positive_num",
    "mall_followers_num",
    "mall_service_score",
    "mall_details_label_list",
    "create_at",
    "update_at",
    "goods_list",
]
import warnings

warnings.filterwarnings("ignore")


class LastDataExport:
    def __init__(self, month=None):
        self._to_db = None

    @property
    def to_db(self):
        if self._to_db is None:
            self._to_db = MysqlDB(
                ip=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_IP"],
                port=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_PORT"],
                db=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_DB"],
                user_name=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_USER_NAME"],
                user_pass=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_USER_PASS"],
            )
        return self._to_db

    @retry(stop_max_attempt_number=5, wait_fixed=5000)
    def find_sql(self, sql):
        return self.to_db.find(sql, to_json=True)

    def get_mall_ids(self):
        return ["392404083"]

    def get_month(self, date_str, days_range=None):
        if days_range is None:
            days_range = [25, 15]
        month = None
        this_date = datetime.datetime.strptime(date_str, "%Y-%m-%d")
        this_month_list = [f"{this_date.strftime('%Y-%m')}-{str(i).zfill(2)}" for i in range(days_range[0], 32)] + [
            f"{(this_date + relativedelta(months=1)).strftime('%Y-%m')}-{str(i).zfill(2)}"
            for i in range(1, days_range[1] + 1)
        ]

        prev_month_list = [f"{this_date.strftime('%Y-%m')}-{str(i).zfill(2)}" for i in range(1, days_range[0])] + [
            f"{(this_date - relativedelta(months=1)).strftime('%Y-%m')}-{str(i).zfill(2)}"
            for i in range(days_range[1], 32)
        ]

        if date_str in this_month_list:
            month = (this_date + relativedelta(months=1)).strftime("%Y%m")
        if date_str in prev_month_list:
            month = this_date.strftime("%Y%m")
        return month

    def deal_single_mall_by_month(self, df, cur_month):
        df_pro = df.copy()
        if df_pro.empty:
            return pd.DataFrame()
        pre_month = (datetime.datetime.strptime(cur_month, "%Y%m") - relativedelta(months=1)).strftime("%Y%m")
        df_pro["_pre_month_date"] = df_pro["month_sale_month_dict"].map(lambda x: x.get(pre_month) if x else None)
        df_pro["_cur_month_date"] = df_pro["month_sale_month_dict"].map(lambda x: x.get(cur_month) if x else None)
        df_pro["_cur_month_sale"] = df_pro["month_sale_trend"].map(lambda x: x.get(cur_month) if x else None)
        df_pro["_cur_month_sale_money"] = df_pro["month_sale_money_trend"].map(
            lambda x: x.get(cur_month) if x else None
        )
        df_pro["_cur_month_sale"] = df_pro["_cur_month_sale"].map(
            lambda x: decimal.Decimal(x) if x is not None else None
        )
        df_pro["_cur_month_sale_money"] = df_pro["_cur_month_sale_money"].map(
            lambda x: decimal.Decimal(x) if x is not None else None
        )

        pre_month_date = df_pro[~df_pro["_pre_month_date"].isna()]["_pre_month_date"].mode().tolist()
        cur_month_date = df_pro[~df_pro["_cur_month_date"].isna()]["_cur_month_date"].mode().tolist()
        mall_id = df_pro["mall_id"].unique()[0]

        if np.all(df_pro["_cur_month_sale"].isna()):
            cur_month_sale = None
            cur_month_sale_money = None
        else:
            cur_month_sale = df_pro["_cur_month_sale"].sum()
            cur_month_sale_money = df_pro["_cur_month_sale_money"].sum()

        if pre_month_date and cur_month_date:
            # cur_month_date_all_list = df_pro[~df_pro["_cur_month_date"].isna()]["_cur_month_date"].unique().tolist()
            # cur_month_date_all_list_str = ','.join([f"'{i}'" for i in cur_month_date_all_list])
            # # 先取合法的version
            # if cur_month_date_all_list:
            #     sql = f"""
            #     select version from source_net_pdd_shop_info where mall_id = '{mall_id}' and version in ({cur_month_date_all_list_str}) and is_strict_all = 1 and (
            #        (source_net_pdd_shop_info.page_goods_count is null and (source_net_pdd_shop_info.crawler_goods_num >= 0))
            #        or
            #        (source_net_pdd_shop_info.page_goods_count and (source_net_pdd_shop_info.crawler_goods_num >= source_net_pdd_shop_info.page_goods_count))
            #     )
            #     """
            #     datas = self.find_sql(sql)
            #     if datas:
            #         cur_month_date = [i['version'] for i in datas]
            cur_month_date = cur_month_date[0]

            cur_month_goods_list = (
                df_pro[~df_pro["_cur_month_date"].isna()][["_cur_month_date", "goods_id", "mall_id"]]
                .rename(columns={"_cur_month_date": "version"})
                .to_dict("records")
            )
            pre_month_goods_list = (
                df_pro[~df_pro["_pre_month_date"].isna()][["_pre_month_date", "goods_id", "mall_id"]]
                .rename(columns={"_pre_month_date": "version"})
                .to_dict("records")
            )

            goods_list = {"cur_month_goods_list": cur_month_goods_list, "pre_month_goods_list": pre_month_goods_list}

            df_pro["_cur_total_sales"] = df_pro["total_sales_month_dict"].map(
                lambda x: (
                    max(list({k: v for k, v in x.items() if v and k <= cur_month}.values()))
                    if list({k: v for k, v in x.items() if v and k <= cur_month}.values())
                    else None
                )
            )
            df_pro["_cur_total_sales_money"] = df_pro["total_sales_money_month_dict"].map(
                lambda x: (
                    max(list({k: v for k, v in x.items() if v and k <= cur_month}.values()))
                    if list({k: v for k, v in x.items() if v and k <= cur_month}.values())
                    else None
                )
            )
            df_pro["_cur_total_sales"] = df_pro["_cur_total_sales"].map(
                lambda x: decimal.Decimal(x) if x is not None else None
            )
            df_pro["_cur_total_sales_money"] = df_pro["_cur_total_sales_money"].map(
                lambda x: decimal.Decimal(x) if x is not None else None
            )

            if np.all(df_pro["_cur_total_sales"].isna()):
                cur_total_sales = None
                cur_total_sales_money = None
            else:
                cur_total_sales = df_pro["_cur_total_sales"].sum()
                cur_total_sales_money = df_pro["_cur_total_sales_money"].sum()

            mall_item = [
                {
                    "mall_id": mall_id,
                    "month_sales": cur_month_sale,
                    "month_sales_money": cur_month_sale_money,
                    "total_sales": cur_total_sales,
                    "total_sales_money": cur_total_sales_money,
                    "version": cur_month_date,
                    "goods_list": goods_list,
                }
            ]
            return pd.DataFrame(mall_item)
        return pd.DataFrame()

    def deal_single_mall(self, df_pro):
        mall_id = df_pro["mall_id"].tolist()[0]
        sql = f"""
        select distinct goods_id, mall_id, max(version) as version
        from source_result_pdd_goods_detail
        where mall_id = '{mall_id}'
        group by goods_id
        """
        # print(sql)
        cur_month_goods_list = self.find_sql(sql)
        pre_month_goods_list = []
        # print(goods_list)
        df_pro["total_sales"] = df_pro["calculate_sales"]
        df_pro["total_sales_money"] = df_pro["calculate_sales_money"]
        df_pro["percent_month_sales"] = None
        df_pro["percent_month_sales_money"] = None
        df_pro["percent_total_sales"] = None
        df_pro["percent_total_sales_money"] = None
        df_pro["goods_list"] = df_pro["mall_id"].map(
            lambda x: {"cur_month_goods_list": cur_month_goods_list, "pre_month_goods_list": pre_month_goods_list}
        )
        df_pro = df_pro.replace({np.NAN: None})
        return df_pro

    def get_mall_datas(self, sqls):
        sql = """select * from source_net_pdd_shop_info where {}"""
        df_all = pd.DataFrame()
        batch_size = 100
        import concurrent.futures

        with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
            tasks = [
                executor.submit(self.find_sql, sql.format(" or ".join(batch_sqls)))
                for batch_sqls in [sqls[i:i + batch_size] for i in range(0, len(sqls), batch_size)]
            ]
            for task in concurrent.futures.as_completed(tasks):
                if task.exception():
                    raise task.exception()
                df_all = pd.concat([df_all, pd.DataFrame(task.result())])
        # datas = self.find_sql(sql.format(" or ".join(sqls)))
        # if datas:
        #     df_all = pd.DataFrame(datas)
        return df_all

    def run_one(self, mall_ids):
        mall_ids_str = ",".join([f"'{i}'" for i in mall_ids])
        sql = f"select * from source_net_pdd_shop_info where mall_id in ({mall_ids_str})"
        datas = self.to_db.find(sql, to_json=True)
        df = pd.DataFrame(datas)
        res_df = df.groupby("mall_id", as_index=False).apply(self.deal_single_mall).reset_index(drop=True)
        res_df = res_df[[i for i in RESULT_MALL_KEYS if i in res_df.columns]]
        save_data = res_df.to_dict("records")
        sqls = []

        if save_data:
            self.to_db.add_batch_smart("zm_result_pdd_shop_info", save_data, update_columns=res_df.columns.tolist())

        # 补充正在上传中的店铺

        for i in save_data:
            mall_id = i.get("mall_id")
            version = i.get("version")
            sqls.append(f" (mall_id = '{mall_id}' and version = '{version}' ) ")

        if sqls:
            new_df = self.get_mall_datas(sqls)
            new_df.rename(
                columns={
                    "shop_name": "mall_name",
                    "shop_url": "mall_url",
                    "shop_status": "mall_status",
                    "service_score": "mall_service_score",
                    "crawler_goods_num": "goods_num",
                },
                inplace=True,
            )
            new_df["mall_open_time_year"] = new_df["open_time_label"].map(lambda x: x.split("年")[0] if x else None)
            new_df["mall_label"] = new_df["mall_label_volist"].map(
                lambda x: ";".join([i["desc"] for i in x]) if x else None
            )
            new_df["mall_achievements_rank"] = new_df["mall_label_volist"].map(
                lambda x: ";".join([i["desc"] for i in x if "榜第" in i["desc"]]) if x else None
            )
            new_df["mall_rating_text"] = new_df["mall_rating_text_list"].map(
                lambda x: (
                    ";".join(
                        [
                            (
                                (i["mallRatingKey"]["txt"] + ":" + i["mallRatingValue"]["txt"])
                                if "mallRatingKey" in i
                                else (i["mall_rating_key"]["txt"] + ":" + i["mall_rating_value"]["txt"])
                            )
                            for i in x
                        ]
                    )
                    if x
                    else None
                )
            )
            result_df = new_df[[i for i in RESULT_MALL_KEYS if i in new_df.columns]]
            result_df.replace({np.nan: None}, inplace=True)
            save_mall_data = result_df.to_dict("records")
            if save_mall_data:
                self.to_db.add_batch_smart(
                    "zm_result_pdd_shop_info", save_mall_data, update_columns=result_df.columns.tolist()
                )

    def run(self, mall_ids):
        batch_size = 10
        futures = []
        with ThreadPoolExecutor(max_workers=10) as executor:
            for i in range(0, len(mall_ids), batch_size):
                mall_ids_batch = mall_ids[i:i + batch_size]
                futures.append(executor.submit(self.run_one, mall_ids_batch))

            for future in tqdm(as_completed(futures), total=len(futures)):
                if future.exception():
                    raise future.exception()
        return []

    def get_goods_datas(self, sqls):
        sql = """select * from source_result_pdd_goods_detail where {}"""
        df_all = pd.DataFrame()
        batch_size = 100

        with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
            tasks = [
                executor.submit(self.find_sql, sql.format(" or ".join(batch_sqls)))
                for batch_sqls in [sqls[i:i + batch_size] for i in range(0, len(sqls), batch_size)]
            ]
            for task in tqdm(concurrent.futures.as_completed(tasks), total=len(tasks), desc=f"任务 pid:{os.getpid()}"):
                if task.exception():
                    raise task.exception()
                df_all = pd.concat([df_all, task.result()])

        return df_all

    def get_month_list(self, month, days_range=None):
        if days_range is None:
            days_range = [25, 15]
        this_date = datetime.datetime.strptime(month, "%Y%m")
        prev_month_list = [f"{this_date.strftime('%Y-%m')}-{str(i).zfill(2)}" for i in range(1, days_range[1] + 1)] + [
            f"{(this_date - relativedelta(months=1)).strftime('%Y-%m')}-{str(i).zfill(2)}"
            for i in range(days_range[0], 32)
        ]
        return prev_month_list

    def get_goods_data(self, params_list):
        batch_size = 100
        datas = []
        with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
            futures = [
                executor.submit(
                    self.find_sql,
                    f"""select version, company_name, mall_name, mall_id, goods_id, goods_name, price, sale_num, level_1, level_2, level_3, goods_rank, hd_url, market_price, normal_price, tag_list from source_result_pdd_goods_detail where { " or ".join(params_list[i:i+batch_size]) }""",
                )
                for i in range(0, len(params_list), batch_size)
            ]
            for future in concurrent.futures.as_completed(futures):
                datas.extend(future.result())
        return pd.DataFrame(datas)

    def export(self, month="202503", mall_ids=["392404083"], version_list=None):
        mall_ids_str = ",".join([f"'{i}'" for i in mall_ids])

        sql = f"""
        SELECT t1.*
        FROM zm_result_pdd_shop_info t1
        JOIN (
            SELECT mall_id, MAX(version) AS max_version
            FROM zm_result_pdd_shop_info
            WHERE mall_id IN ({mall_ids_str})
            GROUP BY mall_id
        ) t2 ON t1.mall_id = t2.mall_id AND t1.version = t2.max_version;
        """
        # sql = f"select * from zm_result_pdd_shop_info where mall_id in ({mall_ids_str}) and version in ({version_list_str})"
        print(sql)
        df_mall = pd.DataFrame(self.to_db.find(sql, to_json=True))
        if df_mall.empty:
            return pd.DataFrame(), pd.DataFrame()

        goods_res = df_mall[~df_mall["goods_list"].isna()]["goods_list"].tolist()

        cur_sqls = []
        pre_sqls = []
        for i in goods_res:
            cur_month_goods_list = i.get("cur_month_goods_list") or []
            pre_month_goods_list = i.get("pre_month_goods_list") or []
            for j in cur_month_goods_list:
                cur_sqls.append(
                    f" ( mall_id = '{j.get('mall_id')}' and goods_id = '{j.get('goods_id')}' and version = '{j.get('version')}' ) "
                )

            for j in pre_month_goods_list:
                pre_sqls.append(
                    f" ( mall_id = '{j.get('mall_id')}' and goods_id = '{j.get('goods_id')}' and version = '{j.get('version')}' ) "
                )

        mall_data_df = df_mall[
            [
                "version",
                "platform_id",
                "platform_name",
                "mall_id",
                "mall_name",
                "company_name_src",
                "taxpayer_id",
                "company_name",
                "month_sales",
                "month_sales_money",
                "total_sales",
                "total_sales_money",
                "percent_month_sales",
                "percent_month_sales_money",
                "percent_total_sales",
                "percent_total_sales_money",
                "goods_num",
                "self_support",
                "mall_url",
                "mall_open_time",
                "mall_open_time_year",
                "mall_status",
                "mall_end_time",
                "mall_platform_classification",
                "mall_business_classification",
                "mall_scope",
                "company_province",
                "company_city",
                "company_county",
                "company_town",
                "mall_location_code",
                "mall_send_province",
                "mall_send_city",
                "mall_send_county",
                "mall_send_location_code",
                "mall_desc",
                "mall_logo",
                "mall_star",
                "mall_label",
                "mall_rating_text",
                "mall_achievements_rank",
                "mall_service_score",
                "mall_details_label_list",
            ]
        ]
        cur_goods_datas_df = pd.DataFrame(self.get_goods_data(cur_sqls))
        pre_goods_datas_df = pd.DataFrame(self.get_goods_data(pre_sqls))

        return mall_data_df, cur_goods_datas_df, pre_goods_datas_df

    def export_mul(self, month):
        mall_ids = self.find_sql("select mall_id from zm_pdd_mall_id where")
        mall_ids_list = [i["mall_id"] for i in mall_ids]
        batch_size = 100
        mall_data_result = pd.DataFrame()
        cur_goods_data_result = pd.DataFrame()
        pre_goods_data_result = pd.DataFrame()
        with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
            futures = [
                executor.submit(self.export, month, mall_ids_list[i:i + batch_size])
                for i in range(0, len(mall_ids_list), batch_size)
            ]
            for future in tqdm(concurrent.futures.as_completed(futures), total=len(futures)):
                if future.exception():
                    raise future.exception()
                mall_data_df, cur_goods_datas_df, pre_goods_datas_df = future.result()
                mall_data_result = pd.concat([mall_data_result, mall_data_df])
                cur_goods_data_result = pd.concat([cur_goods_data_result, cur_goods_datas_df])
                pre_goods_data_result = pd.concat([pre_goods_data_result, pre_goods_datas_df])

        mall_data_result.to_csv(f"export_{month}_mall_data.csv", index=False)
        cur_goods_data_result.to_csv(f"export_{month}_cur_goods_data.csv", index=False)
        pre_goods_data_result.to_csv(f"export_{month}_pre_goods_data.csv", index=False)

    def run_mul(self):
        mall_ids = self.find_sql("select mall_id from zm_pdd_mall_id")
        mall_ids_list = [i["mall_id"] for i in mall_ids]
        batch_size = 100
        with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
            futures = [
                executor.submit(self.run_one, mall_ids_list[i:i + batch_size])
                for i in range(0, len(mall_ids_list), batch_size)
            ]
            for future in tqdm(concurrent.futures.as_completed(futures), total=len(futures)):
                if future.exception():
                    raise future.exception()


if __name__ == "__main__":
    # LastDataExport().run_mul()
    LastDataExport().export_mul("202506")
    # LastDataExport().run()
    # LastDataExport().export_mul("202504")
    # LastDataExport().export_mul("202505")
    # LastDataExport().run_one(['510363394'])

    # 四月份 月销售 8734万8658 月销售额 127亿2671万7097.23
    # 五月份 月销售 5亿9830万1736（包含上月同店铺 3亿0071万3245  新增店铺采集2亿9758万8491 ） 月销售额 775亿3834万1289.7505 （包含上月同店铺 449亿07380639.6001 新增店铺采集326亿3096万0650.1501）
